#include<stdio.h>
#include<pthread.h>
#include<queue>
#include<unistd.h>

using namespace std;

typedef void (*Handler)(int data);
//创建队列处理数据
class QueueData{
    public:
        QueueData(){}
        QueueData(int data, Handler Handler){
            _data = data;
            _handler = Handler;
        }
        void run(){
            _handler(_data);
        }
    private:
        int _data;
        Handler _handler; 
};
#if 1
//线程池
class Threadpool{
    public:
    Threadpool(){}
    Threadpool(int capa, int thread_count){
        _capacity = capa;
        _thread_count = thread_count;
        pthread_mutex_init(&_lock, NULL);
        pthread_cond_init(&_cons_cond, NULL);
        pthread_cond_init(&_prod_cond, NULL);
        flag_exit = 0;
    }
    ~Threadpool(){
        pthread_mutex_destroy(&_lock);
        pthread_cond_destroy(&_cons_cond);
        pthread_cond_destroy(&_prod_cond);
    }
    int OnInit(){
        int count;   //线程创建失败的数量
        for(int i = 0; i < _thread_count; i++){
            pthread_t tid;
            int ret = pthread_create(&tid, NULL, ThreadPollStart, (void*)this);
            if(ret < 0){
                count++;
            }
        }
        return _thread_count -= count;
    }
    void push(QueueData qData){
        pthread_mutex_lock(&_lock);
        while(_que.size() >= _capacity){
            if(flag_exit){
                //告诉生产者别生产了
                return;
            }
            pthread_cond_wait(&_prod_cond, &_lock);
        }
        _que.push(qData);
        pthread_mutex_unlock(&_lock);
        pthread_cond_signal(&_cons_cond);
    }
    void pop(QueueData* qd){
        *qd = _que.front();
        _que.pop();
    }
    //因为线程入口函数只能有一个参数,成员函数会有一个默认的this指针，所以设置为静态函数
    static void* ThreadPollStart(void* arg){
        pthread_detach(pthread_self());
        //要把线程池传进来，也就是this指针
        Threadpool* tp = (Threadpool*)arg;
        while(1){
            pthread_mutex_lock(&tp->_lock);
            while(tp->_que.empty()){
                if(tp->flag_exit){
                    tp->_thread_count--;
                    pthread_mutex_unlock(&tp->_lock);
                    pthread_exit(NULL);
                }
                 pthread_cond_wait(&tp->_cons_cond, &tp->_lock);
            }
            QueueData qd;
            tp->pop(&qd);
            pthread_mutex_unlock(&tp->_lock);
            pthread_cond_signal(&tp->_prod_cond);
            qd.run();
        }
        return NULL;
    }
    //线程退出
    void ThreadpoolExit(){
        flag_exit = 1;
        while(_thread_count){
            //循环通知消费者线程，此时生产者停止了生产，等待队列中可能还有消费者线程
            pthread_cond_signal(&_cons_cond);
        }
    }
    private:
        //一个线程安全队列
        queue<QueueData> _que;
        size_t _capacity;

        pthread_mutex_t _lock;
        pthread_cond_t _cons_cond;
        pthread_cond_t _prod_cond;

        int _thread_count;
        int flag_exit;
};
void DealData(int data){
    printf("data = %d\n", data);
}

#endif
int main(){
    Threadpool* tp = new Threadpool(10, 10);
    if(tp == NULL){
        printf("create fail\n");
        return 0;
    }
    if(tp->OnInit() <= 0){
        printf("create thread fail\n");
        return 0;
    }
    for(int i = 0; i < 10000; i++){
        QueueData qd(i, DealData);
        tp->push(qd);
    }

    tp->ThreadpoolExit();
    delete tp;
    return 0;
}
